package io.plus.interactive;

import drds.common.Clock;
import io.plus.interactive.task.CloseCode;
import io.plus.interactive.user_interface.Packet;
import io.plus.interactive.user_interface.PacketTag;
import lombok.extern.slf4j.Slf4j;

import java.nio.channels.CompletionHandler;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class WriteCompletionHandler implements CompletionHandler<Integer, WriteCompletionVo> {

    public final ReentrantLock lock = new ReentrantLock();
    public final Condition condition = lock.newCondition();
    //
    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;

    public WriteCompletionHandler(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
    }

    @Override
    public void completed(Integer bytesWritten, WriteCompletionVo writeCompletionVo) {
        if (bytesWritten > 0) {
            asynchronousSocketChannelContext.asynchronousSocketChannelState.latestTimeOfSent = Clock.currentTimeMillis;
        }
        if (writeCompletionVo.getByteBuffer().hasRemaining()) {
            if (log.isInfoEnabled()) {
                log.info("{} {}/{} has sent", asynchronousSocketChannelContext, writeCompletionVo.getByteBuffer().position(), writeCompletionVo.getByteBuffer().limit());
            }
            asynchronousSocketChannelContext.asynchronousSocketChannel.write(writeCompletionVo.getByteBuffer(), writeCompletionVo, this);
        } else {
            handlePackets(bytesWritten, null, writeCompletionVo);
        }
    }

    @Override
    public void failed(Throwable throwable, WriteCompletionVo writeCompletionVo) {

        handlePackets(0, throwable, writeCompletionVo);
    }

    public void handlePackets(Integer bytesWritten, Throwable throwable, WriteCompletionVo writeCompletionVo) {
        ReentrantLock lock = asynchronousSocketChannelContext.writeCompletionHandler.lock;
        lock.lock();
        try {
            asynchronousSocketChannelContext.sendTask.canSend = true;
            asynchronousSocketChannelContext.writeCompletionHandler.condition.signal();
            asynchronousSocketChannelContext.asynchronousSocketChannelState.latestTimeOfSentPacket = Clock.currentTimeMillis;
            //


            try {
                boolean isSentSuccess = bytesWritten > 0;
                Object attachment = writeCompletionVo.getObject();
                boolean isPacket = attachment instanceof Packet;
                if (isPacket) {
                    handlePacket((Packet) attachment, isSentSuccess);
                } else {
                    List<?> list = (List<?>) attachment;
                    for (Object object : list) {
                        handlePacket((Packet) object, isSentSuccess);
                    }
                }
                if (!isSentSuccess) {
                    Io.close(asynchronousSocketChannelContext, throwable, "写数据返回:" + bytesWritten, CloseCode.WRITE_COUNT_IS_NEGATIVE);
                }
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }

        } finally {
            lock.unlock();
        }

    }

    public void handlePacket(Packet packet, Boolean isSentSuccess) {
        PacketTag packetTag = packet.getPacketTag();
        if (packetTag != null) {
            packetTag.setIsSentSuccess(isSentSuccess);
        }
        try {
            asynchronousSocketChannelContext.processAfterSent(packet, isSentSuccess);
        } catch (Throwable e) {
            log.error(e.toString(), e);
        }

    }

}
